package com.example.springcloud.rocketmq;

import com.example.springcloud.rocketmq.channel.consume.ConsumeSource;
import com.example.springcloud.rocketmq.channel.produce.ProduceSource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.PollableMessageSource;
import org.springframework.context.annotation.Bean;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;

@SpringBootApplication
@EnableBinding({ProduceSource.class , ConsumeSource.class, SpringcloudAlibabaRocketmqBinderApplication.Source1.class , SpringcloudAlibabaRocketmqBinderApplication.Source2.class})
@Slf4j
public class SpringcloudAlibabaRocketmqBinderApplication {


    public static void main(String[] args) {
        SpringApplication.run(SpringcloudAlibabaRocketmqBinderApplication.class, args);
    }

    @Autowired
    private Source1 source1;

    @Bean
    public CommandLineRunner runner(){
        return (args -> source1.output().send(MessageBuilder.withPayload("paloadStr").setHeader("k1","v1").build()));
    }

    @StreamListener("source1")
    @SendTo("source2")
    public String receiveFromSource1(@Payload String msg) {
        if (log.isInfoEnabled()) {
            log.info("receive message from source1:{}", msg);
        }
        return msg.toUpperCase();
    }

    @StreamListener("source2")
    public void receiveFromSource2(@Payload String msg) {
        if (log.isInfoEnabled()) {
            log.info("receive message from source2:{}", msg);
        }
    }

//   @Bean
//    public ApplicationRunner runner(@Qualifier(value = "pollableMessageSource") PollableMessageSource pollableMessageSource,
//                                    @Qualifier(value = "messageChannel") MessageChannel messageChannel) {
//        return args -> {
//            while (true) {
//                boolean result = pollableMessageSource.poll(m -> {
//                    String payload = (String) m.getPayload();
//                    log.info("Received:{} " , payload);
//                    messageChannel.send(MessageBuilder.withPayload(payload.toUpperCase())
//                            .copyHeaders(m.getHeaders())
//                            .build());
//                }, new ParameterizedTypeReference<String>() { });
//                if (result) {
//                    log.info("Processed a message");
//                }
//                else {
//                    log.info("Nothing to do");
//                }
//                Thread.sleep(5_000);
//            }
//        };
//    }
//    public static interface PolledProcessor {
//        @Input("pollableMessageSource")
//        PollableMessageSource source();
//        @Output("messageChannel")
//        MessageChannel dest();
//    }

    public interface Source1 {
        @Output("source1")
        MessageChannel output();
        @Input("source1")
        SubscribableChannel input();
    }

    public interface Source2 {
        @Output("source2")
        MessageChannel output();
        @Input("source2")
        SubscribableChannel input();
    }


}
